# Copyright (c) 2018 Presto Labs Pte. Ltd.
# Author: leon

import abc
import datetime
from collections import namedtuple
import os
import pathlib

import pandas

from coin.base.datetime_util import to_datetime
from coin.support.feed_tool.feed_stats.logic.util import (datetime_span,
                                                          get_exchange_api_id_from_request)
from coin.base.book.types import (MarketTickerEvent, MarketInstrumentEvent)

Keys = [
    'trading_date',
    'machine',
    'exchange_api_id',
    'native_symbol',
    'instrument_column_name',
    'time_dimension',
    'start_time',
    'end_time'
]

Values = [
    'msg_count',
    'mean',
    'std',
    'min_value',
    'quantile_25',
    'quantile_50',
    'quantile_75',
    'max_value'
]

InstrumentStats = namedtuple('InstrumentStats', Keys + Values)


class InstrumentStatsCalculatorBase(metaclass=abc.ABCMeta):
  def __init__(self, trading_date, machine, subscription_request):
    self._trading_date = trading_date.strftime('%Y%m%d')
    self._machine = machine
    self._sub_request = subscription_request
    self._exchange = subscription_request.exchange
    self._exchange_api_id = \
        get_exchange_api_id_from_request([subscription_request])[0]
    self._instruments = {}
    self._column_names = []
    self._instrument_stats = []

  @property
  def instrument_stats(self):
    return self._instrument_stats

  def set_product(self, product):
    self._instruments[product.native_symbol] = []

  @abc.abstractmethod
  def update_by_instrument(self, instrument):
    raise NotImplementedError()

  def _gen_and_append_stats_row(self, df, start, end, column_name, native_symbol, time_dimension):
    time_fmt = '%Y-%m-%d %H:%M:%S.%f'
    df = df.between_time(start.time(), end.time())
    describe = df[column_name].describe().tolist()
    if len(describe) < len(Values):
      describe = [None] * len(Values)
    start = start.strftime(time_fmt)
    end = end.strftime(time_fmt)
    self._instrument_stats.append(
        InstrumentStats(self._trading_date,
                        self._machine,
                        self._exchange_api_id,
                        native_symbol,
                        column_name,
                        time_dimension,
                        start,
                        end,
                        *describe))

  def hour_stats(self, df, native_symbol):
    start = to_datetime(int(df.iloc[0]['timestamp']))
    end = to_datetime(int(df.iloc[-1]['timestamp']))
    for column_name in self._column_names:
      prev_dt = start
      start_hour = start.hour
      start_hour = start.replace(hour=start_hour + 1, minute=0, second=0, microsecond=0)
      end_hour = end.replace(minute=0, second=0, microsecond=0)
      for dt in datetime_span(start_hour, end_hour, datetime.timedelta(hours=1)):
        self._gen_and_append_stats_row(df, prev_dt, dt, column_name, native_symbol, 'hour')
        prev_dt = dt

      # For the last hour.
      self._gen_and_append_stats_row(df, prev_dt, end, column_name, native_symbol, 'hour')

  def daily_stats(self, df, native_symbol):
    start = to_datetime(int(df.iloc[0]['timestamp']))
    end = to_datetime(int(df.iloc[-1]['timestamp']))
    for column_name in self._column_names:
      self._gen_and_append_stats_row(df, start, end, column_name, native_symbol, 'daily')

  def _pre_process(self):
    for native_symbol in self._instruments.keys():
      df = pandas.DataFrame(self._instruments[native_symbol], columns=self._column_names)
      df['datetime'] = pandas.to_datetime(df['timestamp'], unit='ns')
      df = df.set_index('datetime')
      self._instruments[native_symbol] = df

  def calculator_stats(self):
    self._instrument_stats = []
    for native_symbol in self._instruments.keys():
      df = self._instruments[native_symbol]
      if len(df) == 0:
        continue
      self.hour_stats(df, native_symbol)
      self.daily_stats(df, native_symbol)

  def _generate_csv(self, output_csv_root, instrument_type):
    self._pre_process()
    for native_symbol in self._instruments.keys():
      df = self._instruments[native_symbol]

      dir = os.path.join(output_csv_root, self._machine, self._trading_date)
      pathlib.Path(dir).mkdir(parents=True, exist_ok=True)

      df.to_csv(os.path.join(dir, '%s_%s_%s.csv' %
                             (self._exchange, native_symbol, instrument_type)),
                float_format='%12.6f',
                index=False)


class BitmexInstrumentStatsCalculator(InstrumentStatsCalculatorBase):
  def __init__(self, trading_date, machine, subscription_request):
    super().__init__(trading_date, machine, subscription_request)
    self._column_names = [
        'timestamp',
        'exchange_timestamp',
        'ask_price',
        'bid_price',
        'mid_price',
        'fair_price',
        'fair_basis_rate',
        'funding_rate',
        'indicative_funding_rate',
        'last_change_percent',
        'last_price',
        'mark_price',
        'open_interest',
        'open_value',
        'total_volume',
        'total_turnover',
        'volume24h',
        'turnover24h'
    ]

  def update_by_instrument(self, instrument):
    exchange = instrument.exchange
    assert exchange == 'Bitmex', exchange
    native_symbol = instrument.native_symbol
    assert native_symbol in self._instruments, native_symbol

    timestamp = instrument.timestamp
    instrument = instrument.inst
    row = [
        timestamp,
        instrument['timestamp'],
        instrument['askPrice'],
        instrument['bidPrice'],
        instrument['midPrice'],
        instrument['fairPrice'],
        instrument['fairBasisRate'],
        instrument['fundingRate'],
        instrument['indicativeFundingRate'],
        instrument['lastChangePcnt'],
        instrument['lastPrice'],
        instrument['markPrice'],
        instrument['openInterest'],
        instrument['openValue'],
        instrument['totalVolume'],
        instrument['totalTurnover'],
        instrument['volume24h'],
        instrument['turnover24h']
    ]
    self._instruments[native_symbol].append(row)

  def generate_csv(self, output_csv_root):
    self._generate_csv(output_csv_root, 'instrument')


class OkexInstrumentStatsCalculator(InstrumentStatsCalculatorBase):
  def __init__(self, trading_date, machine, subscription_request):
    super().__init__(trading_date, machine, subscription_request)
    self._column_names = [
        'timestamp',
        'open_interest',
        'volume24h',
    ]

  def update_by_instrument(self, instrument):
    native_symbol = instrument.product.native_symbol
    assert native_symbol in self._instruments, native_symbol

    timestamp = instrument.timestamp
    row = [
        timestamp,
        float(instrument.open_interest),
        float(instrument.vol_24h),
    ]
    self._instruments[native_symbol].append(row)

  def generate_csv(self, output_csv_root):
    self._generate_csv(output_csv_root, 'ticker')


class OkexIndexStatsCalculator(InstrumentStatsCalculatorBase):
  def __init__(self, trading_date, machine, subscription_request):
    super().__init__(trading_date, machine, subscription_request)
    self._column_names = [
        'timestamp',
        'index',
    ]

  def update_by_instrument(self, instrument):
    native_symbol = instrument.product.native_symbol
    assert native_symbol in self._instruments, native_symbol

    timestamp = instrument.timestamp
    row = [
        timestamp,
        float(instrument.index),
    ]
    self._instruments[native_symbol].append(row)

  def generate_csv(self, output_csv_root):
    self._generate_csv(output_csv_root, 'index')


def sample_df_per_seconds(df, seconds):
  return df.resample('%dS' % seconds, closed='left', label='left').ffill()


def gen_df_from_list(instruments, column):
  df = pandas.DataFrame(instruments, columns=['timestamp', column])
  df['datetime'] = pandas.to_datetime(df['timestamp'], unit='ns')
  df = df.set_index('datetime')
  return df


class OkexV3InstrumentStatsCalculator(InstrumentStatsCalculatorBase):
  def __init__(self, trading_date, machine, subscription_request):
    super().__init__(trading_date, machine, subscription_request)
    self._column_names = [
        'timestamp',
        'open_interest',
        'volume24h',
        'mark_price',
    ]
    if self._sub_request.api_version == 'v3_swap':
      self._column_names.append('funding_rate')

  def set_product(self, product):
    self._instruments[product.native_symbol_v3] = {
        'open_interest': [], 'volume24h': [], 'mark_price': [], 'funding_rate': []
    }

  def update_by_instrument(self, instrument):
    native_symbol = instrument.product.native_symbol_v3
    assert native_symbol in self._instruments, native_symbol
    timestamp = instrument.timestamp

    if isinstance(instrument, MarketTickerEvent):
      row = [timestamp, instrument.vol_24h]
      self._instruments[native_symbol]['volume24h'].append(row)
    elif isinstance(instrument, MarketInstrumentEvent):
      if instrument.open_interest is not None:
        row = [timestamp, instrument.open_interest]
        self._instruments[native_symbol]['open_interest'].append(row)
      elif instrument.mark_price is not None:
        row = [timestamp, instrument.mark_price]
        self._instruments[native_symbol]['mark_price'].append(row)
      elif instrument.funding_rate is not None:
        row = [timestamp, instrument.funding_rate]
        self._instruments[native_symbol]['funding_rate'].append(row)
    else:
      raise ValueError(instrument)

  def _pre_process(self):
    for native_symbol in self._instruments.keys():
      open_interest_df = gen_df_from_list(self._instruments[native_symbol]['open_interest'],
                                          'open_interest')
      open_interest_df = sample_df_per_seconds(open_interest_df, 5)
      volume24h_df = gen_df_from_list(self._instruments[native_symbol]['volume24h'], 'volume24h')
      volume24h_df = sample_df_per_seconds(volume24h_df, 5)
      mark_price_df = gen_df_from_list(self._instruments[native_symbol]['mark_price'], 'mark_price')
      mark_price_df = sample_df_per_seconds(mark_price_df, 5)

      df = pandas.merge(open_interest_df,
                        volume24h_df,
                        how='inner',
                        left_index=True,
                        right_index=True)
      df = pandas.merge(df, mark_price_df, how='inner', left_index=True, right_index=True)
      df = df.drop(['timestamp_x', 'timestamp_y'], axis=1)

      if self._sub_request.api_version == 'v3_swap':
        funding_rate_df = gen_df_from_list(self._instruments[native_symbol]['funding_rate'],
                                           'funding_rate')
        funding_rate_df = sample_df_per_seconds(funding_rate_df, 5)
        df = pandas.merge(df, funding_rate_df, how='inner', left_index=True, right_index=True)
        df = df.drop(['timestamp_y'], axis=1)
        df.rename(columns={'timestamp_x': 'timestamp'}, inplace=True)

      df.dropna(how='any', inplace=True)
      self._instruments[native_symbol] = df

  def generate_csv(self, output_csv_root):
    self._generate_csv(output_csv_root, 'instrument')
